package org.dandsoft.blackbox.mapreduce.simple;

import org.dandsoft.blackbox.mapreduce.client.operation.*;
import org.dandsoft.blackbox.mapreduce.data.CollectionValueIterator;
import org.dandsoft.blackbox.mapreduce.processor.ProcessorException;
import org.dandsoft.blackbox.mapreduce.processor.OperationProcessor;
import org.dandsoft.blackbox.mapreduce.processor.OperationProcessorListener;

import java.util.*;

/**
 * @author Vilyuzhanin Denis (Fin)
 */
public class SimpleOperationProcessor implements OperationProcessor {

    private Operation operation;
    private Class<? extends Operation> operationClass;
    private Map<String, String> properties;
    private Map inputMap = new Hashtable();
    private Map outputMap = new Hashtable();
    private Map newOutputMap = new Hashtable();
    private long iterationNumber;
    private Map<Object, LinkedList> temploraryMap = new Hashtable<Object, LinkedList>();
    private Collection<OperationProcessorListener> listeners = new LinkedList<OperationProcessorListener>();

    public void addListener(OperationProcessorListener listener) {
        listeners.add(listener);
    }

    public void removeListener(OperationProcessorListener listener) {
        listeners.remove(listener);
    }

    public void reset() {
        inputMap = new Hashtable();
        outputMap = new Hashtable();
        newOutputMap = new Hashtable();
        temploraryMap = new Hashtable();
        listeners.clear();
        operation = null;
        operationClass = null;
    }

    protected void notifySuccess() {
        for (OperationProcessorListener listener : listeners) {
            listener.success();
        }
    }

    protected void notifyError(Throwable t) {
        for (OperationProcessorListener listener : listeners) {
            listener.error(t);
        }
    }

    public Map getInputMap() {
        return inputMap;
    }

    public Map getOutputMap() {
        return outputMap;
    }

    public Map output() {
        return this.outputMap;
    }

    public void setInputMap(Map inputMap) {
        this.inputMap = inputMap;
    }

    public void input(Map data) {
        this.inputMap.putAll(data);
    }

    public void setOutputMap(Map outputMap) {
        if (outputMap != null) {
            this.outputMap = outputMap;
            this.newOutputMap = outputMap;
        } else {
            this.newOutputMap = new Hashtable();
        }
    }

    public void setOperationClass(Class<? extends Operation> operationClass) {
        this.operationClass = operationClass;

    }

    public Class<? extends Operation> getOperationClass() {
        return operationClass;
    }

    public void setOperation(Operation operation) {
        this.operationClass = operation.getClass();
        this.operation = operation;
    }

    public Operation getOperation() {
        return operation;
    }

    public void setProperties(Map<String, String> properties) {
        this.properties = properties;
    }

    public Map<String, String> getProperties() {
        return properties;
    }

    public void execute() throws ProcessorException {
        if (operation == null) {
            if (operationClass == null) {
                throw new ProcessorException("Operation class is undefine");
            } else {
                operation = OperationFactory.createOperation(operationClass, properties);
            }
        }
        try {
            Context context = new Context();
            operation.init(context);
            executeMap(context);
            executeReduce(context);
            this.outputMap = newOutputMap;
            notifySuccess();
            iterationNumber++;
        } catch (Exception e) {
            notifyError(e);
        }
    }

    /**
     * Выполняет стадию map
     *
     * @param mapContext
     * @throws ProcessorException
     */
    protected void executeMap(Context mapContext) throws ProcessorException {
        try {
            mapContext.setType(PhaseType.Map);
            Set<Map.Entry> entries = inputMap.entrySet();
            for (Map.Entry entry : entries) {
                mapContext.setKey(entry.getKey());
                operation.map(entry.getKey(), entry.getValue());
            }
        } catch (InterruptedException e) {
            throw new ProcessorException(e);
        }
    }

    /**
     * Выполняет стадию reduce
     *
     * @param reduceContext
     * @throws ProcessorException
     */
    protected void executeReduce(Context reduceContext) throws ProcessorException {
        try {
            reduceContext.setType(PhaseType.Reduce);
            for (Map.Entry<Object, LinkedList> entry : temploraryMap.entrySet()) {
                reduceContext.setKey(entry.getKey());
                CollectionValueIterator iterator = new CollectionValueIterator(entry.getValue());
                operation.reduce(entry.getKey(), iterator);
            }
        } catch (InterruptedException e) {
            throw new ProcessorException(e);
        }
    }

    class Context implements OperationContext {
        private PhaseType type;
        private Object key;


        public void emit(Object key, Object value) throws OperationException {
            if (PhaseType.Map != type) {
                throw new WrongEmitOperationException(PhaseType.Map, operation.getOperationName());
            } else {
                LinkedList list = temploraryMap.get(key);
                if (list == null) {
                    list = new LinkedList();
                    temploraryMap.put(key, list);
                }
                list.add(value);
            }
        }

        public void emit(Object value) throws OperationException {
            if (PhaseType.Reduce != type) {
                throw new WrongEmitOperationException(PhaseType.Reduce, operation.getOperationName());
            } else {
                if (newOutputMap.containsKey(getKey())) {
                    throw new KeyAlreadyExistException(operation.getOperationName(), getKey());
                } else {
                    newOutputMap.put(getKey(), value);
                }
            }
        }

        public PhaseType getPhaseType() {
            return type;
        }

        public long getIterationNumber() {
            return iterationNumber;
        }

        public void setType(PhaseType type) {
            this.type = type;
        }

        public Object getKey() {
            return key;
        }

        public void setKey(Object key) {
            this.key = key;
        }
    }

}
